package com.example.kafka.web.controller;

import java.util.concurrent.TimeUnit;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import com.example.kafka.producer.BatchProducer;
import com.example.kafka.producer.Producer;

import lombok.extern.slf4j.Slf4j;

@Controller
@RequestMapping("/test")
@Slf4j
public class SendMessageController {
	
	@Autowired
	private Producer sender;
	
	@Autowired
	private BatchProducer batchSender;

	@RequestMapping("/send")
	@ResponseBody
	public String testReceiver() throws Exception {
		for(int i = 0;i<10000;i++) {
			sender.sendMessage("helloworld.t", i, "Hello Spring Kafka!" + i);
			log.info("send ok");
			TimeUnit.SECONDS.sleep(5);
		}
		return "OK";
	}
	
	@RequestMapping("/batchSend")
	@ResponseBody
	public String batchsend() throws Exception {
		for(int i = 0;i<10000;i++) {
			batchSender.sendMessage("helloworld.t2", i, "Kafka message batch send !" + i);
			log.info("send ok");
		}
		return "OK";
	}

}
